package com.ncmx.lmax.engine;

import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import lombok.extern.log4j.Log4j2;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.ncmx.lmax.engine.essentials.MatchingUnit;
import com.ncmx.lmax.engine.essentials.OrderEntry;

/**
 * Matching Engine will contain 1 or more matching units.
 * 
 * Engine is responsible for listing to incoming requests, journal the requests
 * (orderEntry), replicate to slave Matching engine and then submitting request
 * to Matching Unit for processing
 * 
 * Engine also needs to heat-beat so slave can promote to master if required
 * 
 * Engine will also handle networking - ZeroMQ
 */ 
@Log4j2
public class MatchingEngine {

	/** The Constant RING_SIZE. */
	private final static int RING_SIZE = 1024 * 8;

	/** The executor. */
	private final ExecutorService EXECUTOR = Executors.newFixedThreadPool(3);

	/** The ring buffer. */
	private final RingBuffer<ExchangeEvent> ringBuffer;

	/** The matching unit. */
	//@Autowired
	private  MatchingUnit matchingUnit;

	/** The journal handler. */
	final EventHandler<ExchangeEvent> journalHandler = new EventHandler<ExchangeEvent>() {
		public void onEvent(final ExchangeEvent event, final long sequence,
				final boolean endOfBatch) throws Exception {
			log.info(String.format("Journal %s",event.getValue()));
		}
	};

	/** The replicator handler. */
	final EventHandler<ExchangeEvent> replicatorHandler = new EventHandler<ExchangeEvent>() {
		public void onEvent(final ExchangeEvent event, final long sequence,
				final boolean endOfBatch) throws Exception {
			// log.info(String.format("Replicator %s",
			// event.getValue()));
		}
	};

	/** The matching unit handler. */
	final EventHandler<ExchangeEvent> matchingUnitHandler = new EventHandler<ExchangeEvent>() {
		public void onEvent(final ExchangeEvent event, final long sequence,
				final boolean endOfBatch) throws Exception {
			// log.info(String.format("Process %s",
			// event.getValue()));
			matchingUnit.addOrder(event.getValue());
		}
	};

	/** The disruptor. */
	private final Disruptor<ExchangeEvent> disruptor;

	/**
	 * Instantiates a new matching engine.
	 * 
	 * @param instrument
	 *            the instrument
	 */
	@SuppressWarnings("unchecked")
	public MatchingEngine(final MatchingUnit matchingUnit) {		
		this.matchingUnit=matchingUnit;
		disruptor = new Disruptor<ExchangeEvent>(ExchangeEvent.EVENT_FACTORY,
				RING_SIZE, EXECUTOR, ProducerType.SINGLE,
				new BlockingWaitStrategy());
		disruptor.handleEventsWith(journalHandler, replicatorHandler).then(
				matchingUnitHandler);
		ringBuffer = disruptor.start();
	}

	/**
	 * Publish.
	 * 
	 * @param OrderEntry
	 *            the orderEntry
	 */
	public void publish(final OrderEntry orderEntry) {
		// Publishers claim events in sequence
		long sequence = ringBuffer.next();
		ExchangeEvent event = ringBuffer.get(sequence);
		event.setValue(orderEntry);
		ringBuffer.publish(sequence);
	}

	/**
	 * Halt.
	 */
	protected void halt() {
		disruptor.halt();
		disruptor.shutdown();

	}

	/**
	 * The main method.
	 * 
	 * @param args
	 *            the arguments
	 */
	public static void main(String args[]) {
		//final MatchingEngine engine = new MatchingEngine("MSFT");
		//List<OrderEntry> orderEntryList = new ArrayList<OrderEntry>();
	/*	OrderEntry orderEntry1 = new OrderEntry.Builder(OrderSide.Buy, "B1",
				1234, new PriceLimit(1234)).build();
		orderEntryList.add(orderEntry1);
		OrderEntry orderEntry2 = new OrderEntry.Builder(OrderSide.Buy, "B2",
				1235, new PriceLimit(1235.11)).build();
		orderEntryList.add(orderEntry2);
		OrderEntry orderEntry3 = new OrderEntry.Builder(OrderSide.Sell, "S1",
				1235, new PriceLimit(1235)).build();
		orderEntryList.add(orderEntry3);
		OrderEntry orderEntry4 = new OrderEntry.Builder(OrderSide.Sell, "S2",
				1235, new PriceLimit(1235)).build();
		orderEntryList.add(orderEntry4);
		OrderEntry orderEntry5 = new OrderEntry.Builder(OrderSide.Buy, "B3",
				1235, new PriceLimit(1235.23)).build();
		orderEntryList.add(orderEntry5);
		OrderEntry orderEntry6 = new OrderEntry.Builder(OrderSide.Sell, "S3",
				1234, new PriceLimit(1234.12)).build();
		orderEntryList.add(orderEntry6);*/

//		for (OrderEntry orderEntry : orderEntryList) {
//			engine.publish(orderEntry);
//		}
//
//		log.info("[Enter] q to exit");
//		Scanner scanner = new Scanner(System.in);
//		while (!scanner.nextLine().trim().equals("q")) {
//			log.info("[Enter] q to exit");
//			engine.halt();
//			log.info("Exiting...");
//		}
	}
}
